// 实现异步工作器

#ifndef __M_LOOPER_H__
#define __M_LOOPER_H__

#include <thread>
#include <mutex>
#include <condition_variable>
#include <functional>
#include <memory>
#include <atomic>

#include "buffer.hpp"

namespace superlog
{
    using Functor = std::function<void(Buffer &)>; //函数指针类型

    enum class AsyncType
    {
        ASYNC_SAFE,     //安全状态，表示缓冲区满了则阻塞，避免资源耗尽的风险
        ASYNC_UNSAFE    //不考虑资源耗尽的问题，无限扩容，常用于性能/压力测试(实际不使用)
    };

    class AsyncLooper
    {
    public:
        using ptr = std::shared_ptr<AsyncLooper>;
        AsyncLooper(const Functor &cb, AsyncType looper_type = AsyncType::ASYNC_SAFE) //默认是安全状态
            : _stop(false),
              _looper_type(looper_type),
              _thread(std::thread(&AsyncLooper::threadEntry, this)), //threadEntry()是一个类的成员接口，里面有一个隐藏的this指针
              _callBack(cb)
        {}

        ~AsyncLooper()
        {
            stop(); //停止工作线程
        }

        void stop()
        {
            //_stop这个变量在这里可能在其它线程被调用，该操作存在线程安全问题，所以将其设置为原子类型
            _stop = true; //将退出标志设置为true
            _cond_con.notify_all(); //唤醒所有的工作线程
            _thread.join(); //等待工作线程的退出
        }

        void push(const char* data, size_t len)
        {
            // 1.无限扩容--非安全状态
            // 2.固定大小--生产缓存区中数据满了就阻塞
            std::unique_lock<std::mutex> lock(_mutex);
            // 条件变量控制，若缓冲区剩余空间大小大于数据长度，则可以添加数据，否则阻塞等待
            if(_looper_type == AsyncType::ASYNC_SAFE) //安全状态才需要考虑条件变量的输入控制
            {
                _cond_pro.wait(lock, [&](){
                    return _pro_buf.writeAbleSize() >= len; //被唤醒的条件
                });
            }
            // 此时说明满足条件(缓冲区剩余空间大小>=数据长度)，可以向缓冲区中添加数据
            _pro_buf.push(data, len);
            // 唤醒消费者对缓冲区中的数据进行处理
            _cond_con.notify_one();
        }
    private:
        // 线程(工作线程)入口函数--对消费缓冲区中的数据进行处理(使用回调函数)，处理完毕后，初始化缓冲区，接着交换缓冲区
        void threadEntry()
        {
            while(1)
            {
                // 通过添加一个{}来为互斥锁设置一个生命周期，当缓冲区交换完毕之后就进行解锁(并不对数据的处理过程加锁保护，否则该过程就变成串行化了)
                {
                    // 1.(消费缓冲区最开始肯定是没有数据的)判断生产缓冲区是否有数据，有则交换，无则阻塞
                    std::unique_lock<std::mutex> lock(_mutex);
                    // 退出标志被设置且生产缓冲区已无数据，此时再退出，否则可能会造成生产缓冲区中有数据，但是没有被完全处理的情况
                    if(_stop && _pro_buf.empty())
                    {
                        break;
                    }
                    _cond_con.wait(lock, [&](){
                        // 如果当前是退出前被唤醒，或者是有数据被唤醒，则返回真，继续向下运行，否则重新陷入休眠
                        return _stop || !_pro_buf.empty();
                    });
                    _con_buf.swap(_pro_buf);
                    
                    // 2.只要交换完毕就可以唤醒生产者(生产者可能会因为缓冲区数据满了，导致阻塞，所以需要被唤醒)
                    if(_looper_type == AsyncType::ASYNC_SAFE) //安全状态才需要唤醒，因为该情况下才有可能阻塞
                    {
                        _cond_pro.notify_all();
                    }
                }
                // 3.消费者被唤醒后，对消费缓冲区进行数据处理
                _callBack(_con_buf);
                // 4.初始化消费缓冲区
                _con_buf.reset();
            }
        }
    private:
        Functor _callBack; //具体对消费缓冲区数据进行处理的回调函数，由异步工作器使用者传入
    private:
        AsyncType _looper_type;
        std::atomic<bool> _stop; //工作器的停止标志
        Buffer _pro_buf; //生产缓冲区
        Buffer _con_buf; //消费缓冲区
        std::mutex _mutex; //互斥锁
        std::condition_variable _cond_pro; //生产条件变量
        std::condition_variable _cond_con; //消费条件变量
        std::thread _thread; //异步工作器对应的工作线程
    };
}

#endif